//utf-8
#include "msg.h"
#include "rfmesh.h"


static struct message_t *msg = NULL;


__weak void message_receive_callback(uint8_t id, uint8_t msgsn, int length, void *pdata)
{
	//tr("message(id=%d msgsn=%d total=%d): %.*s\n", \
		id,msgsn,length,length,(char *)pdata);    
}

void message_init(void (*send)(uint8_t,void *))
{
	msg = OS_MALLOC(sizeof(struct message_t));
	INIT_LIST_HEAD(&msg->list);
	for(int i=0;i<MESSAGE_RECORD_COUNT;i++){
		msg->record[i] = 0x0000;
	}
	msg->sn = 0;
	msg->tv = 0;
	msg->send = rfwrite;
	
}



void record_remove_id(uint8_t id)
{
    for(uint8_t i=0;i<MESSAGE_RECORD_COUNT;i++){
        if((msg->record[i] >>8) == id){
            ENTER_SAFE_AREA();
            msg->record[i] = 0;
            EXIT_SAFE_AREA();
        }
    }
}
void record_add(uint8_t id, uint8_t serial)
{
	//tr("record_add() id=0x%x sn=0x%x\n",id,serial);
	
    static uint8_t index = 0;
    ENTER_SAFE_AREA();
		
    msg->record[index] = id<<8 | serial;
    index = (index +1) %MESSAGE_RECORD_COUNT;
    EXIT_SAFE_AREA();
}

int record_check(uint8_t id, uint8_t serial)
{
    int ret = 0;
    uint16_t word = id<<8 | serial;
    for(uint8_t i=0;i<MESSAGE_RECORD_COUNT;i++){
        if(msg->record[i] == word){
            //tr("record exist(id=0x%x sn=0x%x)\n",id,serial);
            ret = 1;
            break;
        }
    }
    return ret;
}



uint32_t absolute(int32_t n)
{ 
	return (n>=0?n:-n);
} 


/*
    发送时把数据封装成 message_srv_t 结构体然后加入链表
    定时轮询链表所有的结构，如果是发送就检查超时并发出对应偏移，如果是接收就检查总的超时并结束
    接收数据处理
    根据记录检查是否接收过，如果是新包就接收、填充结构体加入链表
    遍历所有链表如果符合条件(正在接收中)就写入，注意对方发送的包可能包含上一次offset，做了个判断
    检测接收中的包有没有结束条件，判定是否结束
*/






void update_lock(void)
{
	//in critical
	
	struct list_head *pos;
	struct message_srv_t *obj;
	int local_lock = CONFIG_MSG_LIMIT;
	
	list_for_each(pos,&msg->list){
		obj = list_entry(pos,struct message_srv_t,list);
		
		if(obj->dir == MSG_DIR_RECV)continue;
		if(local_lock <= 0)break;
		obj->lock = MSG_UNLOCK;
		local_lock -= 1;
	}
}

void message_input(uint8_t id, void *pdata)
{

	struct message_data_t *sub = pdata;
	uint8_t MSG_FLAG = sub->flag;
    
	uint8_t FLAG_ACK =   (MSG_FLAG & MSG_FLAG_ACK);
	uint8_t FLAG_END =   (MSG_FLAG & MSG_FLAG_END);
	uint8_t FLAG_START = (MSG_FLAG & MSG_FLAG_START);
    

	if(MSG_FLAG & MSG_FLAG_RST){
		//tr("MSG_FLAG_RST\n");
		record_remove_id(id);//节点重启后第一帧，需要清除之前的MSG_SN
	}

	//如果SN匹配则回复已接收，只有尾帧才会处理；全局使用一个SN池
	//刚上电时，收到对方消息会保存ID+SN，比如2+0，如果此时给对方发消息，则对方会回复之前的ID+SN
	//最终更改： ACK帧检查重复
	if(record_check(id,sub->msgsn) && !FLAG_ACK){
		//tr("record_check\n\n");
		sub->flag = MSG_FLAG_ACK;
		msg->send(id,sub);
		return ;
	}

    //如果空闲就开始接收；验证第一帧
    //tr("id=%d flag=0x%x(START=%d END=%d)\n",msgrx.id,MSG_FLAG,MSG_FLAG & MSG_FLAG_START,MSG_FLAG & MSG_FLAG_END);
    if(MSG_FLAG & MSG_FLAG_START){//需要防止重复，检查是否已经有

        //收到 id==id;再次收到直接写就行，相当于第一个分包被写多次，每次都回应
        struct list_head *cur;
        struct message_srv_t *p;
        list_for_each(cur,&msg->list){
            
            p = list_entry(cur,struct message_srv_t,list);
            if(p->id == id && p->msgsn == sub->msgsn){
                //tr("already start,ack only\n");
                sub->flag = MSG_FLAG_ACK;
								msg->send(id,sub);
                return ;
            }
            
        }

        struct message_srv_t *pmsg = OS_MALLOC(sizeof(struct message_srv_t));
        if(pmsg == NULL) return ;
				
        pmsg->id = id;
        pmsg->dir = MSG_DIR_RECV;
		pmsg->lock = MSG_UNLOCK;
        pmsg->total = sub->total;
        pmsg->pbuf = OS_MALLOC(sub->total);
        pmsg->timebase = msg->tv;
        pmsg->timeout = CONFIG_MSG_DEFAULT_TV;
        
        pmsg->rtt = 0;
        pmsg->rtt_tv = msg->tv;
        pmsg->offset = sub->length;
        pmsg->frame_size = CONFIG_MSG_PACKAGE_SIZE ;
        pmsg->msgsn = sub->msgsn;
        pmsg->pkgsn = sub->pkgsn;
        memcpy(pmsg->pbuf +sub->offset,sub->data,sub->length);
		ENTER_SAFE_AREA();
		list_add_tail(&pmsg->list,&msg->list);
		EXIT_SAFE_AREA();
        tr("new message\n");
        //tr("sub->offset = %d\n",sub->offset);
        //tr("ready recv(next offset=%d len=%d): %.*s\n", \
			pmsg->offset,sub->length,sub->length,pmsg->pbuf);
        //tr("msgrx->pkgsn=%d sub->pkgsn=%d\n",pmsg->pkgsn,sub->pkgsn);
        
        sub->flag = MSG_FLAG_ACK;
		msg->send(id,sub);
    }

    
    struct list_head *cur,*tmp;
    struct message_srv_t *p;
    list_for_each_safe(cur,tmp,&msg->list){
        p = list_entry(cur,struct message_srv_t,list);
        uint8_t ID_OK =      (id == p->id);
        uint8_t SN_OK =      (sub->msgsn == p->msgsn);
        uint8_t SENDER =     (p->dir == MSG_DIR_SEND);
        uint8_t RECVER =     (p->dir == MSG_DIR_RECV);
        
        //收到回复校验信息；自增|结束
        if( SENDER && FLAG_ACK && ID_OK && SN_OK && (p->offset == sub->offset) ){
            
            ENTER_SAFE_AREA();
						
			p->total_retry += (p->retry - 1);
            p->retry = 0;
            p->offset += sub->length;
            p->timebase = 0;
            int rtt = absolute(msg->tv - p->rtt_tv) + 10;
            
            
            //tr("rtt_tv = %d\n",msg_tv);
            
            //计算RTT
            if(p->pkgsn == sub->pkgsn){
                p->rtt = (p->rtt>0)?(rtt + p->rtt)/2:rtt;
                p->timeout = p->rtt * CONFIG_MSG_MULTIPLE_TV;//p->rtt * 3;//发送方的RTT
                //tr("rtt=%d msgrtt=%d msgrto=%d sub->pkgsn=%d tx.pkgsn=%d msgsn=%d\n", \
                rtt,p->rtt,p->timeout,sub->pkgsn,p->pkgsn,p->msgsn);
            }else{
                //tr("# data ok, but pkgsn diff(cur=%d rcv=%d) rtt=%d rto no update\n", \
                p->pkgsn,sub->pkgsn,rtt);
            }
            EXIT_SAFE_AREA();
            
            //判断发送完成
            if(p->offset >= p->total){
                ENTER_SAFE_AREA();
                p->id = 0;
				int total_pakage = p->total / p->frame_size;
				//tr("total: package = %d retry = %d\n",total_pakage, p->total_retry);

                list_del(cur);
				update_lock();
                EXIT_SAFE_AREA();    
                OS_FREE(p->pbuf);
                OS_FREE(p);
                ENTER_SAFE_AREA();
                tr("send ok(msgsn=%d)\n",sub->msgsn);
                EXIT_SAFE_AREA();    
            }
            
            break ;
        }
            
        
        

        
        if( RECVER && ID_OK && SN_OK ){

            //结束标志
            //这里先判断结束，如果结束包里还有数据会被丢弃
            if(FLAG_END){
                ENTER_SAFE_AREA();
                memcpy(p->pbuf +sub->offset,sub->data,sub->length);
                p->offset += sub->length;
                //tr("endpkg recv(next offset=%d len=%d): %.*s\n", \
                p->offset,sub->length,sub->length,p->pbuf +sub->offset);
                EXIT_SAFE_AREA();
                record_add(id,sub->msgsn);
				message_receive_callback(id,sub->msgsn,sub->total,p->pbuf);
                ENTER_SAFE_AREA();
                p->id = 0;
                list_del(cur);
                EXIT_SAFE_AREA();    
                OS_FREE(p->pbuf);
                OS_FREE(p);
                //return ;
            }
            
            //下面的代码仅检测中间分包，不是这两个就是中间分包
            if(FLAG_START || FLAG_END) {
                //tr("FLAG_START || FLAG_END\n");
                break;
            }
                
			tr("msgsn=%d pkgsn=%d p->offset=%d sub->offset=%d\n",p->msgsn,sub->pkgsn,p->offset,sub->offset);
						
            //判断是当前OFFSET还是上一个OFFSET回复，有时ACK后对方没收到就会出现对方重试上一个包，实际本机已经ACK并且OFFSET增加为下一个了
            //或者对方发送的数据超时后才送达
            
            if(sub->offset == p->offset){//4=4
                ENTER_SAFE_AREA();
                memcpy(p->pbuf +sub->offset,sub->data,sub->length);
                p->offset += sub->length;


                
                int rtt = absolute(msg->tv - p->rtt_tv) + 10;
                //tr("msgrx->pkgsn=%d sub->pkgsn=%d\n",p->pkgsn,sub->pkgsn);
                if((p->pkgsn +1) == sub->pkgsn){
                    p->rtt = (p->rtt>0)?(rtt + p->rtt)/2:rtt;
                    p->timeout = \
                    //p->rtt * CONFIG_MSG_MULTIPLE_TV;
                    p->rtt * CONFIG_MSG_MULTIPLE_TV;
                    //uint32_t n = (p->rtt + (p->rtt * CONFIG_MSG_RETRY_COUNT)) * CONFIG_MSG_RETRY_COUNT/2;
                    //tr("rtt = %d timeout=%d\n",p->rtt,p->timeout + n);
                }else{
                    //tr("# pkgsn diff(GOOD=%d rcv=%d) rtt=%d rto no update\n", \
                    (p->pkgsn +1),sub->pkgsn,rtt);
                }
                //tr("rtt=%d msgrtt=%d timeout=%d\n",rtt,p->rtt,p->timeout);
                
                p->timebase = msg->tv;//整个消息超时重新开始
                p->rtt_tv = msg->tv;//单帧RTT测量开始
                p->pkgsn = sub->pkgsn;//记录PKG顺序
                
                EXIT_SAFE_AREA();
                //tr("subpkg recv(next offset=%d len=%d): %.*s\n", \
                p->offset,sub->length,sub->length,p->pbuf +sub->offset);
                
                sub->flag = MSG_FLAG_ACK;
                msg->send(id,sub);

                
            }else
            if(sub->offset == (p->offset - p->frame_size)){//4=0
                ENTER_SAFE_AREA();
                //memcpy(p->pbuf + sub->offset - p->frame_size,sub->data,sub->length);
                
                int rtt = absolute(msg->tv - p->rtt_tv);
                if((p->pkgsn +1) == sub->pkgsn){
                    p->rtt = (p->rtt>0)?(rtt + p->rtt)/2:rtt;
                    p->timeout = \
                    //p->rtt * CONFIG_MSG_MULTIPLE_TV;
                    p->rtt * CONFIG_MSG_MULTIPLE_TV;
                    //tr("rtt = %d timeout=%d\n",p->rtt,p->timeout);
                }else{
                    //tr("# pkgsn diff(GOOD=%d rcv=%d) rtt=%d rto no update\n", \
                    (p->pkgsn +1),sub->pkgsn,rtt);
                }
                //tr("rtt=%d msgrtt=%d timeout=%d\n",rtt,p->rtt,p->timeout);
                
                p->timebase = msg->tv;//整个消息超时重新开始
                p->rtt_tv = msg->tv;//单帧RTT测量开始
                p->pkgsn = sub->pkgsn;//记录PKG顺序
                
                EXIT_SAFE_AREA();
                //tr("# retry pkg: subpkg recv(next offset=%d len=%d): %.*s\n", \
                p->offset,sub->length,sub->length,p->pbuf +sub->offset);
                
                sub->flag = MSG_FLAG_ACK;
                msg->send(id,sub);
                
            }else{
				tr("offset error p->offset=%d sub->offset=%d\n",p->offset,sub->offset);

			}
                
         

        }

    }//list_for_each_safe

	return ;
}




int message_write(uint8_t id, void *pdata, int length)
{

    struct list_head *cur;
    struct message_srv_t *p;
    list_for_each(cur,&msg->list){
        p = list_entry(cur,struct message_srv_t,list);
        if(p->id == id && p->msgsn == (msg->sn+1)){
            //tr("already writed\n");
            return 0;
        }
    }
    
    //类似 FLAG_START
    struct message_srv_t *pmsg = OS_MALLOC(sizeof(struct message_srv_t));
    if(pmsg == NULL) return 0;
    
    memset(pmsg,0,sizeof(struct message_srv_t));
    pmsg->dir = MSG_DIR_SEND;
    pmsg->id = id;
    pmsg->pbuf = OS_MALLOC(length);
    pmsg->offset = 0;
    pmsg->total = length;
    pmsg->frame_size = CONFIG_MSG_PACKAGE_SIZE;
    pmsg->msgsn = msg->sn ++;
    pmsg->timebase = 0;
    pmsg->retry = 0;
    pmsg->pkgsn = 0;
    pmsg->rtt = 0;
	pmsg->lock = MSG_LOCKED;
	pmsg->timeout = CONFIG_MSG_DEFAULT_TV;
	memcpy(pmsg->pbuf,pdata,length);
	ENTER_SAFE_AREA();
	list_add_tail(&pmsg->list,&msg->list);
	update_lock();
	EXIT_SAFE_AREA();
	tr("message_write_to_server(msgsn=%d)\n",pmsg->msgsn);

    return 1;
}

void message_server(void)
{
    struct list_head *cur,*tmp;
    struct message_srv_t *p;
    
    msg->tv += 10;
    
    list_for_each_safe(cur,tmp,&msg->list){
        p = list_entry(cur,struct message_srv_t,list);
			
        if(p->id && p->dir == MSG_DIR_RECV){//dir=recv
            
            //接收任务仅计算超时退出
            uint32_t n = (p->rtt + (p->rtt * CONFIG_MSG_RETRY_COUNT)) * CONFIG_MSG_RETRY_COUNT/2;
            if((msg->tv - p->timebase) > (p->timeout + n)){
                ENTER_SAFE_AREA();
                p->timebase = msg->tv;
                p->id = 0;
                list_del(cur);
                EXIT_SAFE_AREA();
                OS_FREE(p->pbuf);
                OS_FREE(p);
                tr("rx timeout(msgsn=%d)\n",p->msgsn);
            }            
        }
        

        
        if( p->id && p->dir == MSG_DIR_SEND   && p->lock ==MSG_UNLOCK  ){
            if((msg->tv - p->timebase) > (p->timeout + (p->rtt * p->retry))){
                
                //发送任务超时重试，超时退出
                ENTER_SAFE_AREA();
                p->timebase = msg->tv;
                p->retry ++;
                //if(p->retry >1)tr("retry = %d,rtt = %d timeout = %d\n", \
                    p->retry,p->rtt,p->timeout + p->rtt * p->retry);
                EXIT_SAFE_AREA();
                
                if(p->retry >= CONFIG_MSG_RETRY_COUNT){
                    ENTER_SAFE_AREA();
                    list_del(cur);
                    p->id = 0;
					update_lock();
                    EXIT_SAFE_AREA();
                    OS_FREE(p->pbuf);
                    OS_FREE(p);
										
                    tr("send cancel(msgsn=%d)\n",p->msgsn);
                    return ;
                }
                
                struct message_data_t *sub = OS_MALLOC(sizeof(struct message_data_t));
                memset(sub,0,sizeof(struct message_data_t));

                int send_bytes = 0;
                int free_bytes = p->total - p->offset;//剩余字节
                int len =  (free_bytes > p->frame_size)?p->frame_size : free_bytes;
                
                if(p->offset == 0){
                    sub->flag |= MSG_FLAG_START;
					//这里不能完全解决问题 如果在重启前给多个设备发送消息，重启后只会给第一个设备发送复位
                    static uint8_t static_reset_label = 0;
                    if(!static_reset_label /* && FIRST */){
                        //tr("MSG_FLAG_RST\n");
                        static_reset_label = 1;
                        sub->flag |= MSG_FLAG_RST;
                    }
                    tr("MSG_FLAG_START(msgsn=%d)\n",p->msgsn);
                }
                if(p->offset + len == p->total) {
                    sub->flag |= MSG_FLAG_END;
                    //tr("MSG_FLAG_END\n");
                }else
                if(p->offset != 0 && free_bytes > 0) {
                    sub->flag = 0 ;
                }

                
                ENTER_SAFE_AREA();
                memcpy(sub->data,p->pbuf + (p->offset),len);
                sub->total = p->total;
                sub->length = len;
                sub->msgsn = p->msgsn;
                sub->offset = p->offset;
                sub->pkgsn = ++p->pkgsn;
                p->rtt_tv = msg->tv;
                //tr("rtt_tv = %d\n",msg_tv);
                //tr("rfwrite(retry=%d msgsn=%d pkgsn=%d offset=%d)\n", \
					p->retry,sub->msgsn,sub->pkgsn,sub->offset);
                //tr("rfwrite(retry=%d len=%d msgsn=%d pkgsn=%d rto=%d flag=0x%02x)\n", \
					p->retry,sub->length,sub->msgsn,sub->pkgsn,p->timeout,sub->flag);
                EXIT_SAFE_AREA();
                msg->send(p->id,sub);
                OS_FREE(sub);
            }
        }
				
    }

}





